package org.hbhk.hms.mq.tx.consumer;

import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.hbhk.hms.mq.tx.mq.JsonUtil;
import org.hbhk.hms.mq.tx.mq.MqMsgVo;
import org.hbhk.hms.mq.tx.provider.IMqRecordService;
import org.hbhk.hms.mq.tx.provider.MqRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.env.Environment;

/**
 * {MIN_OFFSET=0, MAX_OFFSET=16, KEYS=121212, CONSUME_START_TIME=1585183458954,
 * UNIQ_KEY=C0A8CB57317C18B4AAC2829F027B0000, CLUSTER=DefaultCluster, WAIT=true,
 * TAGS=*, key=121212}
 * 
 * @author Administrator
 *
 * @param <T>
 */
public abstract class MqMessageListener<T> implements IMqtxMessageListener<T>, InitializingBean {

	private Logger log = LoggerFactory.getLogger(getClass());

	@Autowired
	private IMqRecordService mqRecordService;

	@Autowired
	private Environment environment;

	@Value("${hms.mq.namesrvAddr}")
	private String namesrvAddr;

	private ExecutorService executor = Executors.newCachedThreadPool();

	private ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(1);

	public abstract void onMessage(Map<String, String> header, T message);

	@Override
	public void onMessage(MqMsgVo<T> mqMsgVo) {
		Map<String, String> header = mqMsgVo.getHeader();
		T message = mqMsgVo.getBody();
		try {
			onMessage(header, message);
		} catch (Exception e) {
			executor.execute(new Runnable() {
				@Override
				public void run() {
					try {
						MqConsumer consumer = this.getClass().getAnnotation(MqConsumer.class);
						MqRequest mqRequest = new MqRequest();
						mqRequest.setTopic("mq_fail_record");
						mqRequest.setTags(consumer.groupName());
						mqRequest.setMsgKey(mqMsgVo.getKey());
						MqFailRecord mqFailRecord = new MqFailRecord();
						mqFailRecord.setMsgId(mqMsgVo.get("msgId"));
						mqFailRecord.setExceptionMsg(ExceptionUtils.getStackTrace(e));
						mqFailRecord.setGroup(consumer.groupName());
						mqFailRecord.setTopic(consumer.topic());
						mqFailRecord.setKey(mqMsgVo.getKey());
						mqFailRecord.setParam(JsonUtil.objectToJson(message));
						mqRequest.setBody(JsonUtil.objectToJson(mqFailRecord));
						mqRecordService.sendMessage(mqRequest);
					} catch (Exception e2) {
						log.error("发现失败消息异常:" + e2.getMessage());
					}
				}
			});

			throw new RuntimeException(e);
		}
	}

	
	@Override
	public void afterPropertiesSet() throws Exception {

		MqConsumer consumer = this.getClass().getAnnotation(MqConsumer.class);

		String groupName = resolve(consumer.groupName());

		if (ConsumerContainer.listenerMap.containsKey(groupName)) {
			throw new RuntimeException("消费组重复:" + groupName);
		}

		IMqtxMessageListener<T> thisClass = this;

		ConsumerConfig config = new ConsumerConfig();
		String namesrvAddr = consumer.namesrvAddr();
		if (StringUtils.isEmpty(namesrvAddr)) {
			namesrvAddr = this.namesrvAddr;
		} else {
			namesrvAddr = resolve(namesrvAddr);
		}
		config.setGroup(groupName);
		config.setConsumeMode(consumer.consumeMode());
		config.setInstanceName(consumer.instanceName());
		config.setServerUrl(namesrvAddr);
		config.setTag(consumer.tags());
		config.setThreadMax(consumer.threadMax());
		config.setThreadMin(consumer.threadMin());
		config.setTopic(consumer.topic());
		config.setUnitName(consumer.unitName());
		config.setVipChannelEnabled(consumer.vipChannelEnabled());
		ConsumerContainer.addConsumer(config, thisClass);

		scheduledExecutor.scheduleAtFixedRate(new Runnable() {

			@Override
			public void run() {
				try {
					log.debug("加载mq配置");
					String threadNum = environment.getProperty(groupName + ".threadNum");
					String start = environment.getProperty(groupName + ".start");
					Boolean isStart = true;
					try {
						if (StringUtils.isEmpty(start)) {
							return;
						}
						isStart = Boolean.valueOf(start);
					} catch (Exception e) {
						return;
					}
					if (StringUtils.isNotEmpty(start) && isStart == false) {
						ConsumerContainer.stop(groupName);
						log.error("消费组:" + groupName + "停止");
						return;
					}
					if (StringUtils.isNotEmpty(threadNum)) {
						int threadMax = ConsumerContainer.listenerThreadNum.get(groupName);
						int threadNumConfig = Integer.parseInt(threadNum);
						
						if (threadMax != threadNumConfig) {
							config.setThreadMax(threadNumConfig);
							int threadMin = threadNumConfig / 2;
							if (threadMin == 0) {
								threadMin = 1;
							}
							config.setThreadMin(threadMin);
							ConsumerContainer.stop(groupName);
							log.error("消费组:" + groupName + "停止");
							ConsumerContainer.addConsumer(config, thisClass);
							log.error("消费组:" + groupName + "启动");
						}
					}

				} catch (Exception e) {
					log.error("消费组配置错误：" + e.getMessage());
				}

			}
		}, 0, 60, TimeUnit.SECONDS);
	}

	private String resolve(String value) {
		if (org.springframework.util.StringUtils.hasText(value)) {
			return this.environment.resolvePlaceholders(value);
		}
		return value;
	}

}
